package utils

import (
	"fmt"
	"sync"
	"time"

	"boomlive.com/boomlive-common/configs"
	"github.com/beego/beego/v2/core/logs"
	amqp "github.com/rabbitmq/amqp091-go"
)

type RabbitMq struct {
	conn           *amqp.Connection
	publishChannel *amqp.Channel
	declaredQueues map[string]bool
	mu             sync.Mutex
}

var rabbitMqClient *RabbitMq
var rabbitMqClientOnce sync.Once

func GetRabbitMq() *RabbitMq {
	rabbitMqClientOnce.Do(func() {
		rabbitMqClient = &RabbitMq{}
	})
	return rabbitMqClient
}

func (r *RabbitMq) getConn() *amqp.Connection {
	if r.conn == nil || r.conn.IsClosed() {
		r.mu.Lock()
		if r.conn == nil || r.conn.IsClosed() {
			var err error
			r.conn, err = amqp.Dial(configs.MqConnStr)
			if err != nil {
				logs.Error("failed to connect to RabbitMQ: %s", err)
			} else {
				logs.Info("connected to RabbitMQ")
			}
		}
		r.mu.Unlock()
	}
	return r.conn
}

func (r *RabbitMq) resetConn() {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.conn != nil {
		_ = r.conn.Close()
		r.conn = nil
		r.publishChannel = nil
		r.declaredQueues = make(map[string]bool)
	}
}

func (r *RabbitMq) getChannel() *amqp.Channel {
	var conn = r.getConn()
	if conn == nil {
		return nil
	}
	var err error
	channel, err := conn.Channel()
	if err != nil {
		r.resetConn()
		logs.Error("failed to open a channel: err=%s, connect is reset", err)
		return nil
	}

	return channel
}

func (r *RabbitMq) declareQueue(ch *amqp.Channel, routingKey string) error {
	if declared := r.declaredQueues[routingKey]; declared {
		return nil
	}
	_, err := ch.QueueDeclare(routingKey, true, false, false, false, nil)
	if err != nil {
		logs.Error("failed to declare queue %s: err=%s", routingKey, err)
		return err
	}
	return nil
}

func (r *RabbitMq) getPublishChannel() *amqp.Channel {
	if r.publishChannel == nil || r.publishChannel.IsClosed() {
		r.publishChannel = r.getChannel()
	}
	return r.publishChannel
}

func (r *RabbitMq) Publish(routingKey string, msg string) error {
	ch := r.getPublishChannel()
	if ch == nil {
		return fmt.Errorf("failed to get publish channel, routingKey=%s msg=%s", routingKey, msg)
	}

	err := r.declareQueue(ch, routingKey)
	if err != nil {
		return fmt.Errorf("failed to declare queue, routingKey=%s msg=%s error=%s", routingKey, msg, err.Error())
	}

	err = ch.Publish("", routingKey, false, false, amqp.Publishing{
		ContentType: "text/plain",
		Body:        []byte(msg),
	})
	if err != nil {
		logs.Error("Failed to publish message: %s", err)
		return err
	}

	return nil
}

func (r *RabbitMq) Listen(routingKey string, listener func(msg string)) error {
	// 创建一个channel
	ch := r.getChannel()
	if ch == nil {
		return fmt.Errorf("failed to get channel, routingKey=%s", routingKey)
	}

	err := r.declareQueue(ch, routingKey)
	if err != nil {
		return fmt.Errorf("failed to declare queue, routingKey=%s error=%s", routingKey, err.Error())
	}

	// 创建消费者
	msgs, err := ch.Consume(
		routingKey, // 消费消息的队列名,从这个队列中接收消息
		"",         // 消费者的标签，通常是消费者的标识。可以使用这个字段来标识不同的消费者
		true,       // 是否启用自动确认。如果为 true，消息一旦被消费就会自动确认
		false,      // 是否为排他性消费者。true 表示只有当前消费者可以使用该队列
		false,      // 是否禁止消费者接收自己发送到队列的消息
		false,      // 是否等待服务器的确认。当设置为 true 时，RabbitMQ 会返回响应但不会等待确认队列存在等其他信息。false 则表示需要等待队列的确认（通常是检查队列是否已经创建好）
		nil,        // 消费者的其他可选参数
	)
	if err != nil {
		return fmt.Errorf("failed to create consume, routingKey=%s, err=%s", routingKey, err)
	}

	go func() {
		defer func() {
			if panicErr := recover(); panicErr != nil {
				logs.Error("%s Listen is done: %s", routingKey, PanicTrace(panicErr))
			}
			//Retry after 2 seconds
			timer := time.NewTimer(2 * time.Second)
			<-timer.C

			go r.Listen(routingKey, listener)

		}()

		for d := range msgs {
			logs.Info(string(d.Body))
			listener(string(d.Body))
		}
	}()

	logs.Info("[*] Listening %s", routingKey)
	return nil
}
